package edu.nepu.flink.api.transform;

import edu.nepu.flink.api.bean.WaterSensor;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @Date 2024/2/28 20:41
 * @Created by chenshuaijun
 */
public class FlatmapOperator {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        /**
         * flatmap的特点是：一进多出
         * 可以是
         *  一进一出
         *  一进多出
         *  一进零出
         * 所以flatmap的应用是非常的广泛的，因为他是非常的灵活
         * flatmap之所以可以实现这种效果主要是因为，数据的输出调用的是Collection，调用几次就会返回多少次数据
         *
         * map之所以必须是一进一出，主要的原因是map使用的是return来输出数据，所以必须要返回，并且只能返回一个参数。
         */

        DataStreamSource<String> streamSource = env.socketTextStream("hadoop102", 9999);

        SingleOutputStreamOperator<String> flatMapStream = streamSource.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                String[] words = value.split(" ");
                /*
                输出的方式：
                （1）如果输入的是hello，我们就输出hello world
                （2）如果输入的是flink，就输出两个flink
                （3）对于其它的情况我们不进行输出
                 */
                for (String word : words) {
                    if ("hello".equals(word)) {
                        out.collect("hello world");
                    } else if ("flink".equals(word)) {
                        out.collect("flink");
                        out.collect("flink");
                    }
                }
            }
        });

        flatMapStream.print();

        env.execute();
    }
}
